package org.example.timer.connectionpool;

import lombok.extern.slf4j.Slf4j;

import java.util.ArrayDeque;
import java.util.concurrent.atomic.AtomicLong;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.ReentrantLock;

/**
 * 模拟数据库连接池
 */
@Slf4j
public class CustomDataSourcePool {
    private ReentrantLock lock = new ReentrantLock();

    private Condition empty = lock.newCondition();
    private Condition notEmpty = lock.newCondition();

    /**
     * 资源队列
     */
    private ArrayDeque<String> queue = new ArrayDeque<>();

    private final int maxSize = 300;

    private final AtomicLong pollingCount = new AtomicLong(0);

    private final AtomicLong waitCount = new AtomicLong(0);

    private final AtomicLong activeCount = new AtomicLong(0);

    private final AtomicLong appleId = new AtomicLong(0);

    private volatile boolean inited = false;

    private Thread createThread;

    public CustomDataSourcePool() {
        createThread = new Thread(new Producter());
        createThread.start();
    }

    public void init() {
        lock.lock();
        try {
            if (inited)
                return;
            log.info("初始化链接");
            inited = true;
        } finally {
            lock.unlock();
        }
    }

    /**
     * 生产者线程
     */
    public class Producter implements Runnable {
        @Override
        public void run() {
            // 负责创建链接
            for (; ; ) {
                lock.lock();
                try {
                    // 说明资源足够，不需要生产
                    if (pollingCount.get() >= waitCount.get()) {
                        try {
                            empty.await();
                        } catch (InterruptedException e) {
                            e.printStackTrace();
                        }
                    }
                    // 如果队列已经满了，空了，才能往下走
                    if (queue.size() == maxSize) {
                        try {
                            empty.await();
                        } catch (InterruptedException e) {
                            e.printStackTrace();
                        }
                    }

                    // 生产消息
                    String connection = getDirectConnection();
                    log.info("生产了：" + connection + "，通知其他消费者");
                    queue.addLast(connection);
                    // 通知所有等待的人，池子有资源了
                    notEmpty.signalAll();
                    // 池子数量+1
                    pollingCount.incrementAndGet();
                } finally {
                    lock.unlock();
                }
            }
        }
    }

    /**
     * 获取链接
     *
     * @return
     */
    public String getConnection() {
        init();
        String result = null;
        lock.lock();
        try {
            // 没有连接数量了
            while (pollingCount.get() == 0) {
                // 通知生产者该生产数据了
                log.info("资源没有了，通知生产者生产");
                empty.signal();

                // 继续等待
                try {
                    waitCount.incrementAndGet();
                    notEmpty.await();
                    waitCount.decrementAndGet();
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }

            }
            result = queue.pollFirst();
            activeCount.incrementAndGet();
            // 池子数量减1
            pollingCount.decrementAndGet();
            return result;
        } finally {
            lock.unlock();
        }
    }

    /**
     * 直接获取物理链接
     *
     * @return
     */
    public String getDirectConnection() {
        return new String("苹果" + appleId.incrementAndGet());
    }

    public long getPollingCount() {
        return pollingCount.get();
    }
}
